//! SQLite存储实现

use async_trait::async_trait;
use nacos_core::{Storage, Result, ConfigStorage, ServiceStorage, ConfigItem, ServiceInstance, ConfigType};
use sqlx::sqlite::{SqlitePool, SqlitePoolOptions};
use std::collections::HashMap;
use crate::error::StorageError;
use tracing::{debug, trace, info, warn};
use chrono::{DateTime, Utc};

/// SQLite存储实现
#[derive(Debug)]
pub struct SqliteStorage {
    pool: SqlitePool,
}

impl SqliteStorage {
    /// 创建新的SQLite存储实例
    pub async fn new(connection_string: &str) -> std::result::Result<Self, sqlx::Error> {
        info!("Initializing SQLite storage with connection: {}", connection_string);
        let pool = SqlitePoolOptions::new()
            .max_connections(5)
            .connect(connection_string).await?;
        
        // 初始化表结构
        Self::init_tables(&pool).await?;

        Ok(Self { pool })
    }

    /// 初始化数据库表
    async fn init_tables(pool: &SqlitePool) -> std::result::Result<(), sqlx::Error> {
        debug!("Initializing database tables for SQLite");
        
        // 通用KV表
        sqlx::query(
            "CREATE TABLE IF NOT EXISTS kv (
                key TEXT PRIMARY KEY,
                value BLOB NOT NULL
            );"
        ).execute(pool).await?;
        
        // Configs table (SQLite specific schema)
        sqlx::query(
            "CREATE TABLE IF NOT EXISTS configs (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                namespace TEXT NOT NULL,
                group_name TEXT NOT NULL,
                data_id TEXT NOT NULL,
                content TEXT,
                config_type TEXT,
                md5 TEXT NOT NULL,
                version INTEGER NOT NULL DEFAULT 1,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                creator TEXT,
                description TEXT,
                UNIQUE(namespace, group_name, data_id)
            );"
        ).execute(pool).await?;

        // Service instances table (SQLite specific schema)
        sqlx::query(
            "CREATE TABLE IF NOT EXISTS service_instances (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                instance_id TEXT NOT NULL,
                service_name TEXT NOT NULL,
                group_name TEXT NOT NULL,
                namespace TEXT NOT NULL,
                ip TEXT NOT NULL,
                port INTEGER NOT NULL,
                weight REAL NOT NULL,
                healthy INTEGER NOT NULL, -- Using INTEGER for BOOLEAN
                enabled INTEGER NOT NULL,
                ephemeral INTEGER NOT NULL,
                cluster_name TEXT NOT NULL,
                metadata TEXT,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                UNIQUE(namespace, group_name, service_name, instance_id)
            );"
        ).execute(pool).await?;

        debug!("Database tables initialized successfully");
        Ok(())
    }
}

#[async_trait]
impl Storage for SqliteStorage {
    async fn get(&self, key: &str) -> Result<Option<Vec<u8>>> {
        trace!("Getting key from SQLite: {}", key);
        let row: Option<(Vec<u8>,)> = sqlx::query_as("SELECT value FROM kv WHERE key = ?")
            .bind(key)
            .fetch_optional(&self.pool)
            .await
            .map_err(|e| StorageError::Database(e.to_string()))?;
        Ok(row.map(|r| r.0))
    }

    async fn put(&self, key: &str, value: &[u8]) -> Result<()> {
        trace!("Putting key to SQLite: {}", key);
        sqlx::query("INSERT OR REPLACE INTO kv (key, value) VALUES (?, ?)")
            .bind(key)
            .bind(value)
            .execute(&self.pool)
            .await
            .map_err(|e| StorageError::Database(e.to_string()))?;
        Ok(())
    }

    async fn delete(&self, key: &str) -> Result<()> {
        trace!("Deleting key from SQLite: {}", key);
        sqlx::query("DELETE FROM kv WHERE key = ?")
            .bind(key)
            .execute(&self.pool)
            .await
            .map_err(|e| StorageError::Database(e.to_string()))?;
        Ok(())
    }

    async fn scan(&self, prefix: &str) -> Result<Vec<(String, Vec<u8>)>> {
        trace!("Scanning SQLite with prefix: {}", prefix);
        let rows: Vec<(String, Vec<u8>)> = sqlx::query_as("SELECT key, value FROM kv WHERE key LIKE ?")
            .bind(format!("{}%", prefix))
            .fetch_all(&self.pool)
            .await
            .map_err(|e| StorageError::Database(e.to_string()))?;
        Ok(rows)
    }
    
    async fn batch_get(&self, keys: &[String]) -> Result<HashMap<String, Vec<u8>>> {
        warn!("batch_get for SQLite is not implemented efficiently and falls back to iterative gets.");
        let mut result = HashMap::new();
        for key in keys {
            if let Some(value) = self.get(key).await? {
                result.insert(key.clone(), value);
            }
        }
        Ok(result)
    }
    
    async fn batch_put(&self, data: &HashMap<String, Vec<u8>>) -> Result<()> {
        warn!("batch_put for SQLite is not implemented efficiently and falls back to iterative puts.");
        for (key, value) in data {
            self.put(key, value).await?;
        }
        Ok(())
    }
    
    async fn batch_delete(&self, keys: &[String]) -> Result<()> {
        warn!("batch_delete for SQLite is not implemented efficiently and falls back to iterative deletes.");
        for key in keys {
            self.delete(key).await?;
        }
        Ok(())
    }

    async fn exists(&self, key: &str) -> Result<bool> {
        let row: Option<(i64,)> = sqlx::query_as("SELECT 1 FROM kv WHERE key = ? LIMIT 1")
            .bind(key)
            .fetch_optional(&self.pool)
            .await
            .map_err(|e| StorageError::Database(e.to_string()))?;
        Ok(row.is_some())
    }

    async fn keys(&self, pattern: &str) -> Result<Vec<String>> {
        let rows: Vec<(String,)> = sqlx::query_as("SELECT key FROM kv WHERE key LIKE ?")
            .bind(pattern.replace('*', "%"))
            .fetch_all(&self.pool)
            .await
            .map_err(|e| StorageError::Database(e.to_string()))?;
        Ok(rows.into_iter().map(|r| r.0).collect())
    }

    async fn clear(&self) -> Result<()> {
        sqlx::query("DELETE FROM kv")
            .execute(&self.pool)
            .await
            .map_err(|e| StorageError::Database(e.to_string()))?;
        Ok(())
    }
}

#[async_trait]
impl ConfigStorage for SqliteStorage {
    async fn get_config(&self, namespace: &str, group: &str, data_id: &str) -> Result<Option<ConfigItem>> {
        let row: Option<ConfigRow> = sqlx::query_as(
            "SELECT * FROM configs WHERE namespace = ? AND group_name = ? AND data_id = ?"
        )
        .bind(namespace).bind(group).bind(data_id)
        .fetch_optional(&self.pool).await
        .map_err(|e| StorageError::Database(e.to_string()))?;
        
        Ok(row.map(|r| r.into()))
    }
    
    async fn save_config(&self, config: &ConfigItem) -> Result<()> {
        sqlx::query(
            "INSERT INTO configs (namespace, group_name, data_id, content, config_type, md5, version, created_at, updated_at, creator, description) \
             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \
             ON CONFLICT(namespace, group_name, data_id) DO UPDATE SET \
             content=excluded.content, config_type=excluded.config_type, md5=excluded.md5, version=excluded.version, updated_at=excluded.updated_at"
        )
        .bind(&config.namespace)
        .bind(&config.group)
        .bind(&config.data_id)
        .bind(&config.content)
        .bind(serde_json::to_string(&config.config_type).unwrap_or_default())
        .bind(&config.md5)
        .bind(config.version as i64)
        .bind(config.create_time.to_rfc3339())
        .bind(config.update_time.to_rfc3339())
        .bind(&config.creator)
        .bind(&config.description)
        .execute(&self.pool).await
        .map_err(|e| StorageError::Database(e.to_string()))?;
        
        Ok(())
    }
    
    async fn delete_config(&self, namespace: &str, group: &str, data_id: &str) -> Result<()> {
        sqlx::query("DELETE FROM configs WHERE namespace = ? AND group_name = ? AND data_id = ?")
            .bind(namespace).bind(group).bind(data_id)
            .execute(&self.pool).await
            .map_err(|e| StorageError::Database(e.to_string()))?;
        Ok(())
    }
    
    async fn list_configs(&self, namespace: &str, group: Option<&str>) -> Result<Vec<ConfigItem>> {
        let mut query = "SELECT * FROM configs WHERE namespace = ?".to_string();
        if group.is_some() {
            query.push_str(" AND group_name = ?");
        }

        let mut query_builder = sqlx::query_as::<_, ConfigRow>(&query).bind(namespace);
        if let Some(g) = group {
            query_builder = query_builder.bind(g);
        }

        let rows = query_builder.fetch_all(&self.pool).await.map_err(|e| StorageError::Database(e.to_string()))?;
        Ok(rows.into_iter().map(|r| r.into()).collect())
    }
    
    async fn get_config_history(&self, _namespace: &str, _group: &str, _data_id: &str, _limit: usize) -> Result<Vec<ConfigItem>> {
        // SQLite implementation does not have a separate history table in this basic version.
        // Returning the current version as a history of one.
        warn!("get_config_history for SQLite returns only the current version.");
        if let Some(config) = self.get_config(_namespace, _group, _data_id).await? {
            Ok(vec![config])
        } else {
            Ok(vec![])
        }
    }
}

#[async_trait]
impl ServiceStorage for SqliteStorage {
    async fn register_instance(&self, instance: &ServiceInstance) -> Result<()> {
        sqlx::query(
            "INSERT INTO service_instances (namespace, group_name, service_name, instance_id, ip, port, weight, healthy, enabled, ephemeral, cluster_name, metadata, created_at, updated_at) \
             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \
             ON CONFLICT(namespace, group_name, service_name, instance_id) DO UPDATE SET \
             ip=excluded.ip, port=excluded.port, weight=excluded.weight, healthy=excluded.healthy, enabled=excluded.enabled, ephemeral=excluded.ephemeral, cluster_name=excluded.cluster_name, metadata=excluded.metadata, updated_at=excluded.updated_at"
        )
        .bind(&instance.namespace)
        .bind(&instance.group_name)
        .bind(&instance.service_name)
        .bind(&instance.instance_id)
        .bind(&instance.ip)
        .bind(instance.port as i64)
        .bind(instance.weight)
        .bind(instance.healthy)
        .bind(instance.enabled)
        .bind(instance.ephemeral)
        .bind(&instance.cluster_name)
        .bind(serde_json::to_string(&instance.metadata).unwrap_or_default())
        .bind(instance.create_time.to_rfc3339())
        .bind(instance.update_time.to_rfc3339())
        .execute(&self.pool).await
        .map_err(|e| StorageError::Database(e.to_string()))?;
        Ok(())
    }
    
    async fn deregister_instance(&self, namespace: &str, group: &str, service_name: &str, instance_id: &str) -> Result<()> {
        sqlx::query("DELETE FROM service_instances WHERE namespace = ? AND group_name = ? AND service_name = ? AND instance_id = ?")
            .bind(namespace).bind(group).bind(service_name).bind(instance_id)
            .execute(&self.pool).await
            .map_err(|e| StorageError::Database(e.to_string()))?;
        Ok(())
    }
    
    async fn get_instance(&self, namespace: &str, group: &str, service_name: &str, instance_id: &str) -> Result<Option<ServiceInstance>> {
        let row: Option<InstanceRow> = sqlx::query_as(
            "SELECT * FROM service_instances WHERE namespace = ? AND group_name = ? AND service_name = ? AND instance_id = ?"
        )
        .bind(namespace).bind(group).bind(service_name).bind(instance_id)
        .fetch_optional(&self.pool).await
        .map_err(|e| StorageError::Database(e.to_string()))?;
        
        Ok(row.map(|r| r.into()))
    }
    
    async fn get_all_instances(&self, namespace: &str, group: &str, service_name: &str) -> Result<Vec<ServiceInstance>> {
        let rows: Vec<InstanceRow> = sqlx::query_as(
            "SELECT * FROM service_instances WHERE namespace = ? AND group_name = ? AND service_name = ?"
        )
        .bind(namespace).bind(group).bind(service_name)
        .fetch_all(&self.pool).await
        .map_err(|e| StorageError::Database(e.to_string()))?;
        
        Ok(rows.into_iter().map(|r| r.into()).collect())
    }
    
    async fn update_instance_health(&self, namespace: &str, group: &str, service_name: &str, instance_id: &str, healthy: bool) -> Result<()> {
        sqlx::query(
            "UPDATE service_instances SET healthy = ?, updated_at = ? WHERE namespace = ? AND group_name = ? AND service_name = ? AND instance_id = ?"
        )
        .bind(healthy)
        .bind(Utc::now().to_rfc3339())
        .bind(namespace).bind(group).bind(service_name).bind(instance_id)
        .execute(&self.pool).await
        .map_err(|e| StorageError::Database(e.to_string()))?;
        Ok(())
    }
    
    async fn get_all_services(&self, namespace: &str, group: Option<&str>) -> Result<Vec<String>> {
        let mut query = "SELECT DISTINCT service_name FROM service_instances WHERE namespace = ?".to_string();
        if group.is_some() {
            query.push_str(" AND group_name = ?");
        }

        let mut query_builder = sqlx::query_as::<_, (String,)>(&query).bind(namespace);
        if let Some(g) = group {
            query_builder = query_builder.bind(g);
        }

        let rows = query_builder.fetch_all(&self.pool).await.map_err(|e| StorageError::Database(e.to_string()))?;
        Ok(rows.into_iter().map(|r| r.0).collect())
    }
}

// Helper structs for mapping from sqlx rows

#[derive(sqlx::FromRow)]
struct ConfigRow {
    namespace: String,
    group_name: String,
    data_id: String,
    content: Option<String>,
    config_type: Option<String>,
    md5: String,
    version: i64,
    created_at: String,
    updated_at: String,
    creator: Option<String>,
    description: Option<String>,
}

impl From<ConfigRow> for ConfigItem {
    fn from(row: ConfigRow) -> Self {
        Self {
            namespace: row.namespace,
            group: row.group_name,
            data_id: row.data_id,
            content: row.content.unwrap_or_default(),
            config_type: serde_json::from_str(&row.config_type.unwrap_or_default()).unwrap_or(ConfigType::Text),
            md5: row.md5,
            version: row.version as u64,
            create_time: DateTime::parse_from_rfc3339(&row.created_at).unwrap_or_default().with_timezone(&Utc),
            update_time: DateTime::parse_from_rfc3339(&row.updated_at).unwrap_or_default().with_timezone(&Utc),
            creator: row.creator,
            description: row.description,
        }
    }
}

#[derive(sqlx::FromRow)]
struct InstanceRow {
    namespace: String,
    group_name: String,
    service_name: String,
    instance_id: String,
    ip: String,
    port: i64,
    weight: f64,
    healthy: bool,
    enabled: bool,
    ephemeral: bool,
    cluster_name: String,
    metadata: String,
    created_at: String,
    updated_at: String,
}

impl From<InstanceRow> for ServiceInstance {
    fn from(row: InstanceRow) -> Self {
        Self {
            namespace: row.namespace,
            group_name: row.group_name,
            service_name: row.service_name,
            instance_id: row.instance_id,
            ip: row.ip,
            port: row.port as u16,
            weight: row.weight,
            healthy: row.healthy,
            enabled: row.enabled,
            ephemeral: row.ephemeral,
            cluster_name: row.cluster_name,
            metadata: serde_json::from_str(&row.metadata).unwrap_or_default(),
            create_time: DateTime::parse_from_rfc3339(&row.created_at).unwrap_or_default().with_timezone(&Utc),
            update_time: DateTime::parse_from_rfc3339(&row.updated_at).unwrap_or_default().with_timezone(&Utc),
        }
    }
}
